use std::{
    collections::{HashMap, HashSet},
    fs::File,
    path::Path,
    sync::{atomic::AtomicBool, Arc},
};

use common::{
    base::{config::Config, dxc::Manifest, id_generator, status::Status},
    serial::api_descriptor::ServiceApiDescriptor,
    service::{
        sys_service_api::{
            self,
            xport_service::{ServiceInfo, ServiceInfos},
        },
        ServiceKey,
    },
    utils::file_utils,
};
use semver::Version;
use tokio::{sync::RwLock, task};
use tracing::info;

use crate::service::{
    loader::DllService,
    local_service::LocalService,
    remote_service::RemoteService,
    service::Service,
    sys_service_api::xport_service::{local_service_off, local_service_on},
};

use super::airport::Airport;

struct LocalServiceHolder {
    service_id_map: HashMap<i64, Arc<Box<LocalService>>>,
    service_name_map: HashMap<String, Arc<Box<LocalService>>>,
}

struct RemoteServiceHolder {
    // 连接信息对应的 channel
    service_conn_id_map: HashMap<i64, HashSet<i64>>,
    service_id_map: HashMap<i64, Arc<Box<RemoteService>>>,
    service_name_map: HashMap<String, Arc<Box<RemoteService>>>,
}

pub struct DXC {
    // /**
    //  * 所有 service 的实例名
    //  */
    // _service_name_set: RwLock<HashSet<String>>,
    dxc_name: RwLock<String>,
    /**
     *
     */
    airport: Arc<Box<Airport>>,
    /**
     *
     */
    is_close: AtomicBool,
    /**
     *
     */
    local_service_holder: RwLock<LocalServiceHolder>,
    /**
     *
     */
    remote_service_holder: RwLock<RemoteServiceHolder>,

    /**
     * 版本
     */
    version: Version,
    /**
     *
     */
    dxc_id: i64,
}

impl DXC {
    pub fn new(airport: Arc<Box<Airport>>, version: Version) -> Self {
        DXC {
            // _service_name_set: RwLock::new(HashSet::new()),
            airport: airport,
            dxc_name: RwLock::new(String::default()),
            is_close: AtomicBool::new(false),
            local_service_holder: RwLock::new(LocalServiceHolder {
                service_id_map: HashMap::new(),
                service_name_map: HashMap::new(),
            }),
            remote_service_holder: RwLock::new(RemoteServiceHolder {
                service_conn_id_map: HashMap::new(),
                service_id_map: HashMap::new(),
                service_name_map: HashMap::new(),
            }),
            version: version,
            dxc_id: id_generator::gen_dxc_id(),
        }
    }

    pub async fn set_name(&self, name: String) {
        let mut self_name = self.dxc_name.write().await;
        *self_name = name;
    }

    pub async fn name(&self) -> String {
        let name = self.dxc_name.read().await;
        name.clone()
    }

    pub fn is_close(&self) -> bool {
        self.is_close.load(std::sync::atomic::Ordering::Relaxed)
    }

    pub fn set_close(&self) -> bool {
        let ret = self.is_close.compare_exchange(
            false,
            true,
            std::sync::atomic::Ordering::AcqRel,
            std::sync::atomic::Ordering::Acquire,
        );
        ret.is_ok()
    }
    /**
     *
     */
    pub fn verison(&self) -> Version {
        self.version.clone()
    }
    /**
     *
     */
    pub fn id(&self) -> i64 {
        self.dxc_id
    }

    pub async fn load_local_service(&self, service_name: &str, config: &str) -> Result<(), Status> {
        let dxc_name = self.name().await;
        let dxc_version = self.verison();
        let version_with_line = dxc_version.to_string().replace(".", "_");
        let dxc_name_with_version: String = format!("{}_{}", dxc_name, version_with_line);
        let dxc_unzip_path = format!("./dxc/{}", dxc_name_with_version);

        let dxc_path = format!("./dxc/{}.dxc", dxc_name_with_version);

        if !file_utils::is_file_exist(&dxc_path).await {
            return Err(Status::error(format!("加载失败: dxc :{} 不存在", dxc_path)));
        }

        let dxc_md5 = file_utils::get_md5(&dxc_path);

        if dxc_md5.is_err() {
            return Err(Status::error(format!(
                "加载失败: dxc :{} 不存在",
                dxc_md5.err().unwrap().to_string()
            )));
        }
      

        let dxc_md5 = dxc_md5.unwrap();
        if !file_utils::is_dir_exist(&dxc_unzip_path).await {
            file_utils::create_dir(&dxc_unzip_path).await;
            let input = File::open(&dxc_path)
                .expect(&format!("打开文件失败，请检查文件{}是否存在", &dxc_path));

            let zip = zip::ZipArchive::new(input);
            if zip.is_err() {
                return Err(Status::error(format!(
                    "打开文件失败，请检查文件:{} 是否是 DXC 文件",
                    dxc_path
                )));
            }
            let mut zip = zip.unwrap();
            for i in 0..zip.len() {
                let file = zip.by_index(i); //.expect("读取文件失败！");
                if file.is_err() {
                    return Err(Status::error(format!(
                        "读取DXC:{} 失败",
                        &dxc_name_with_version
                    )));
                }
                let mut file = file.unwrap();
                let file_name = file.mangled_name();
                // 创建文件的完整路径
                let output_path = Path::new(&dxc_unzip_path).join(file_name);
                let output_file = File::create(&output_path);
                if output_file.is_err() {
                    return Err(Status::error(format!(
                        "创建 DXC: {} 文件 {} 失败",
                        &dxc_name_with_version,
                        &output_path.to_str().unwrap()
                    )));
                }

                let mut output_file = output_file.unwrap();
                let ret = std::io::copy(&mut file, &mut output_file);
                if ret.is_err() {
                    return Err(Status::error(format!(
                        "写入 DXC: {}  文件 {} 失败",
                        &dxc_name_with_version,
                        &output_path.to_str().unwrap()
                    )));
                };
            }
        }

        let manifest_path = format!("{}/manifest.json", &dxc_unzip_path);

        let manifest = file_utils::read_file_as_string(&manifest_path).await;

        if manifest.is_err() {
            return Err(Status::error(format!(
                "加载DXC:{} manifest.json文件失败，请检查dxc是否合法",
                &dxc_name_with_version
            )));
        }

        let manifest = manifest.unwrap();

        let manifest = serde_json::from_str::<Manifest>(&manifest);

        if manifest.is_err() {
            return Err(Status::error(format!(
                "读取DXC:{} manifest.json文件失败，请检查dxc是否合法",
                &dxc_name_with_version
            )));
        }
        //
        let manifest = manifest.unwrap();

        if manifest.system != std::env::consts::OS {
            return Err(Status::error(format!(
                "加载DXC:{} 失败,当前系统为：{}, 不能加载 {} 的DXC",
                &dxc_name_with_version,
                std::env::consts::OS,
                &manifest.system
            )));
        }
        //
        let dsl_path = format!("{}/source_api.proto", dxc_unzip_path);

        let api_descriptor =
            ServiceApiDescriptor::load_with_include_path(&dsl_path, &dxc_unzip_path, true);

        if api_descriptor.is_err() {
            return Err(Status::error(format!(
                "加载DXC:{} 描述文件失败，请检查dxc是否合法",
                dxc_name_with_version
            )));
        }

        let api_descriptor = api_descriptor.unwrap();

        let dxc_path = if std::env::consts::OS == "windows" {
            format!("{}/{}.dll", &dxc_unzip_path, &dxc_name)
        } else {
            format!("{}/{}.so", &dxc_unzip_path, &dxc_name)
        };

        let dll_service = DllService::new(&dxc_path);

        if dll_service.is_err() {
            return Err(Status::error(format!(
                "加载DXC:{} 应用失败，请检查dxc是否合法",
                dxc_name_with_version
            )));
        }

        let dll_service = dll_service.unwrap();

        let service_config = Config::load_from_string(config);

        let is_system = service_config.get_bool("", "system").unwrap_or(false);

        let local_service = LocalService::new(
            dxc_name,
            dxc_version,
            self.dxc_id,
            dxc_md5,
            is_system,
            service_name.into(),
            self.airport.clone(),
            dll_service,
            api_descriptor,
        );

        let local_service = Arc::new(local_service);
        {
            let mut local_service_hoder = self.local_service_holder.write().await;

            local_service_hoder
                .service_id_map
                .insert(local_service.id(), local_service.clone());

            local_service_hoder
                .service_name_map
                .insert(local_service.name().into(), local_service.clone());
        }

        local_service.start();

        let config = config.to_string();
        let clone_local_service = local_service.clone();
        let _ = task::spawn_blocking(move || {
            clone_local_service.init(config.as_str());
        })
        .await;

        let service_info = local_service.get_service_info();

        let airport = local_service.airport.clone();

        tokio::spawn(async move {
            local_service_on(airport, service_info).await;
        });

        Ok(())
    }

    pub async fn remove_local_service_by_name(&self, service_name: &str) {
        let local_service = {
            let mut local_service_holder = self.local_service_holder.write().await;

            let local_service = local_service_holder.service_name_map.remove(service_name);

            if local_service.is_none() {
                return;
            }
            let local_service = local_service.unwrap();

            local_service_holder
                .service_id_map
                .remove(&local_service.id());

            local_service
        };

        info!("移除 {} 服务", local_service.name());

        let clone_local_service = local_service.clone();
        let _ = task::spawn_blocking(move || {
            clone_local_service.finalize();
        })
        .await;

        local_service.stop().await;
        let service_info = local_service.get_service_info();
        let airport = local_service.airport.clone();

        drop(local_service);
        tokio::spawn(async move {
            local_service_off(airport, service_info).await;
        });
    }
    /**
     *
     */
    pub async fn get_local_service_by_id(&self, service_id: i64) -> Option<Arc<Box<LocalService>>> {
        let local_service_holder = self.local_service_holder.read().await;

        let local_service = local_service_holder.service_id_map.get(&service_id);

        return if let Some(local_service) = local_service {
            Some(local_service.clone())
        } else {
            None
        };
    }
    /**
     *
     */
    pub async fn get_local_service_by_name(
        &self,
        service_name: &str,
    ) -> Option<Arc<Box<LocalService>>> {
        let local_service_holder = self.local_service_holder.read().await;

        let local_service = local_service_holder.service_name_map.get(service_name);

        return if let Some(local_service) = local_service {
            Some(local_service.clone())
        } else {
            None
        };
    }
    /**
     *
     */
    pub async fn get_remote_service_by_id(
        &self,
        service_id: i64,
    ) -> Option<Arc<Box<RemoteService>>> {
        let remote_service_holder = self.remote_service_holder.read().await;

        let remote_service = remote_service_holder.service_id_map.get(&service_id);

        return if let Some(remote_service) = remote_service {
            Some(remote_service.clone())
        } else {
            None
        };
    }
    /**
     *
     */
    pub async fn get_remote_service_by_name(
        &self,
        service_name: &str,
    ) -> Option<Arc<Box<RemoteService>>> {
        let remote_service_holder = self.remote_service_holder.read().await;
        let remote_service = remote_service_holder.service_name_map.get(service_name);
        return if let Some(remote_service) = remote_service {
            Some(remote_service.clone())
        } else {
            None
        };
    }
    /**
     *
     */
    pub async fn get_remote_service_by_conn_id(
        &self,
        conn_id: i64,
    ) -> Option<Arc<Box<RemoteService>>> {
        let remote_service_holder = self.remote_service_holder.write().await;
        let id_set = remote_service_holder.service_conn_id_map.get(&conn_id);
        if id_set.is_none() {
            return None;
        }
        //
        for service_id in id_set.unwrap().iter() {
            let remote_service = remote_service_holder.service_id_map.get(service_id);
            if remote_service.is_none() {
                continue;
            }
            Some(remote_service.unwrap().clone());
        }
        None
    }

    /**
     *
     */
    pub async fn get_service_key(&self, service_id: i64) -> ServiceKey {
        let local_service = self.get_local_service_by_id(service_id).await;
        if local_service.is_some() {
            return local_service.unwrap().get_service_key();
        }
        let remote_service = self.get_remote_service_by_id(service_id).await;
        if remote_service.is_some() {
            return remote_service.unwrap().get_service_key();
        }
        ServiceKey::default()
    }
    /**
     *
     */
    pub async fn get_service_id_by_name(&self, service_name: &str) -> i64 {
        {
            let local_service_holder = self.local_service_holder.read().await;
            let local_service = local_service_holder.service_name_map.get(service_name);
            let local_service = local_service.unwrap();
            let service_id = local_service.id();
            if service_id != 0 {
                return service_id;
            }
        }

        let remote_service_holder = self.remote_service_holder.read().await;
        let remote_service = remote_service_holder.service_name_map.get(service_name);
        let remote_service = remote_service.unwrap();
        remote_service.id()
    }

    /**
     *
     */
    pub async fn get_service_by_id(&self, service_id: i64) -> Option<Arc<Box<dyn Service>>> {
        let local_service = self.get_local_service_by_id(service_id).await;
        if let Some(local_service) = local_service {
            return Some(Arc::downcast(local_service).unwrap());
        }
        //
        let remote_service = self.get_remote_service_by_id(service_id).await;
        if let Some(remote_service) = remote_service {
            return Some(Arc::downcast(remote_service).unwrap());
        }
        //
        None
    }

    /**
     *
     */
    pub async fn get_service_name_by_id(&self, service_id: i64) -> Option<String> {
        let service = self.get_service_by_id(service_id).await;
        return if let Some(service) = service {
            Some(String::from(service.name()))
        } else {
            None
        };
    }
    /**
     *
     */
    pub async fn has_local_service_by_id(&self, service_id: i64) -> bool {
        let local_service = self.get_local_service_by_id(service_id).await;

        local_service.is_some()
    }
    /**
     *
     */
    pub async fn has_local_service_by_name(&self, service_name: &str) -> bool {
        let local_service = self.get_local_service_by_name(service_name).await;

        local_service.is_some()
    }

    pub async fn has_local_service(&self) -> bool {
        let local_service_holder = self.local_service_holder.read().await;
        local_service_holder.service_id_map.len() != 0
    }
    /**
     *
     */
    pub async fn has_remote_service_by_id(&self, service_id: i64) -> bool {
        let remote_service = self.get_remote_service_by_id(service_id).await;

        remote_service.is_some()
    }
    /**
     *
     */
    pub async fn has_remote_service_by_name(&self, service_name: &str) -> bool {
        let remote_service = self.get_remote_service_by_name(service_name).await;

        remote_service.is_some()
    }

    /**
     *
     */
    pub async fn remove_service_by_id(&self, service_id: i64) -> bool {
        {
            let mut local_service_holder = self.local_service_holder.write().await;

            let local_service = local_service_holder.service_id_map.remove(&service_id);

            if let Some(local_service) = local_service {
                local_service_holder
                    .service_name_map
                    .remove(local_service.name());
                local_service.finalize();
                local_service.stop().await;
                return true;
            }
        }

        self.remove_remote_service_by_id(service_id).await
    }

    async fn remove_remote_service_by_id(&self, service_id: i64) -> bool {
        let mut remote_service_holder = self.remote_service_holder.write().await;
        let remote_service = remote_service_holder.service_id_map.remove(&service_id);
        if let Some(remote_service) = remote_service {
            remote_service_holder
                .service_name_map
                .remove(remote_service.name());

            self.remove_remote_service_id_in_conn_map(service_id).await;

            remote_service.finalize();
            remote_service.stop().await;
            return true;
        }
        false
    }
    /**
     *
     */
    async fn remove_remote_service_by_name(&self, service_name: &str) -> bool {
        let mut remote_service_holder = self.remote_service_holder.write().await;
        let remote_service = remote_service_holder.service_name_map.remove(service_name);
        if let Some(remote_service) = remote_service {
            let service_id = remote_service.id();
            remote_service_holder.service_id_map.remove(&service_id);

            self.remove_remote_service_id_in_conn_map(service_id).await;

            remote_service.finalize();
            remote_service.stop().await;
            return true;
        }
        false
    }

    pub async fn remove_service_by_name(&self, service_name: &str) -> bool {
        {
            let mut local_service_holder = self.local_service_holder.write().await;
            let local_service = local_service_holder.service_name_map.remove(service_name);
            if let Some(local_service) = local_service {
                let service_id = local_service.id();

                local_service_holder.service_id_map.remove(&service_id);
                local_service.finalize();
                local_service.stop().await;
                return true;
            }
        }

        self.remove_remote_service_by_name(service_name).await
    }

    pub async fn ack_message(&self, ack_message_id: i64) -> bool {
        let remote_service_holder = self.remote_service_holder.read().await;
        for (_, remote_service) in remote_service_holder.service_id_map.iter() {
            if remote_service.ack_message(ack_message_id).await {
                return true;
            }
        }
        false
    }

    pub async fn add_remote_service_and_return_ins(
        &self,
        service_name: &str,
    ) -> Arc<Box<RemoteService>> {
        let mut remote_service_holder = self.remote_service_holder.write().await;

        let remote_service = remote_service_holder.service_name_map.get(service_name);

        if remote_service.is_some() {
            return remote_service.unwrap().clone();
        }

        let remote_service = RemoteService::new(
            self.dxc_id,
            self.name().await,
            self.verison(),
            String::from(service_name),
            self.airport.clone(),
        );

        let remote_service = Arc::new(remote_service);

        remote_service_holder
            .service_id_map
            .insert(remote_service.id(), remote_service.clone());

        remote_service_holder
            .service_name_map
            .insert(String::from(remote_service.name()), remote_service.clone());

        let clone_remote_service = remote_service.clone();

        // 启动接收消息

        clone_remote_service.start();

        remote_service
    }

    /**
     * 添加 remote service
     */
    pub async fn add_remote_service(&self, channel_id: i64, service_name: &str) {
        let remote_service = self.add_remote_service_and_return_ins(service_name).await;
        remote_service.add_channel_id(channel_id);
    }
    /**
     *
     */
    pub async fn remote_service_has_channel(&self, service_id: i64) -> bool {
        let remote_service = self.get_remote_service_by_id(service_id).await;
        if remote_service.is_none() {
            return false;
        }
        let remote_service = remote_service.unwrap();
        remote_service.has_channel()
    }

    pub async fn add_channel_to_remote_service_by_conn_id(&self, conn_id: i64, channel_id: i64) {
        let remote_service_holder = self.remote_service_holder.write().await;

        let id_set = remote_service_holder.service_conn_id_map.get(&conn_id);

        if id_set.is_none() {
            return;
        }
        for service_id in id_set.unwrap().iter() {
            let remote_service = remote_service_holder.service_id_map.get(service_id);
            if remote_service.is_none() {
                continue;
            }
            remote_service.unwrap().add_channel_id(channel_id);
        }
    }

    pub async fn remove_channel_in_remote_service_by_conn_id(&self, conn_id: i64, channel_id: i64) {
        let remote_service_holder = self.remote_service_holder.write().await;
        let id_set = remote_service_holder.service_conn_id_map.get(&conn_id);
        if id_set.is_none() {
            return;
        }
        for service_id in id_set.unwrap().iter() {
            let remote_service = remote_service_holder.service_id_map.get(service_id);
            if remote_service.is_none() {
                continue;
            }
            remote_service.unwrap().remove_channel_id(channel_id);
        }
    }

    /**
     *
     */
    pub async fn add_channel_to_remote_service(&self, service_name: &str, channel_id: i64) {
        let remote_service = self.add_remote_service_and_return_ins(service_name).await;

        remote_service.add_channel_id(channel_id);

        // 添加进去
        self.add_remote_service_id_to_conn_map(channel_id, remote_service.id())
            .await;
    }

    async fn add_remote_service_id_to_conn_map(&self, channel_id: i64, service_id: i64) {
        let channel = self
            .airport
            .get_channel_manager()
            .get_channel(channel_id)
            .await;

        if channel.is_none() {
            return;
        }
        let channel = channel.unwrap();

        let mut remote_service_holder = self.remote_service_holder.write().await;
        let id_set = remote_service_holder
            .service_conn_id_map
            .entry(channel.conn_id())
            .or_insert(HashSet::new());

        id_set.insert(service_id);
    }

    async fn remove_remote_service_id_in_conn_map(&self, service_id: i64) {
        let mut remote_service_holder = self.remote_service_holder.write().await;
        for (_, id_set) in remote_service_holder.service_conn_id_map.iter_mut() {
            if id_set.remove(&service_id) {
                return;
            }
        }
    }

    /**
     *
     */
    pub async fn update_remote_service_info(
        &self,
        channel_id: i64,
        service_name: &str,
        service_id: i64,
    ) {
        let remote_service = self.add_remote_service_and_return_ins(service_name).await;
        // 这里要删除

        let cur_service_id = remote_service.id();
        // 删除即可
        // 在添加
        self.remove_remote_service_id_in_conn_map(cur_service_id)
            .await;

        remote_service.add_channel_id(channel_id);

        remote_service.update_id(service_id);
    }
    /**
     *  移除远程Service的通道信息
     */
    pub async fn remove_all_remote_service_channel_id(&self, channel_id: i64) {
        //
        let remote_service_holder = self.remote_service_holder.read().await;

        for (_key, value) in remote_service_holder.service_id_map.iter() {
            value.remove_channel_id(channel_id);
        }
    }
    /**
     *
     */
    pub async fn message_in_by_id(&self, service_id: i64, message: Arc<Vec<u8>>) -> bool {
        let service = self.get_service_by_id(service_id).await;

        if service.is_none() {
            return false;
        }

        let service = service.unwrap();

        service.message_in(message.as_ref())
    }
    /**
     *
     */
    pub async fn message_in_by_name(&self, service_name: &str, message: &Vec<u8>) -> bool {
        let local_service = self.get_local_service_by_name(service_name).await;

        if let Some(local_service) = local_service {
            local_service.message_in(message);

            return true;
        }

        let remote_service = self.add_remote_service_and_return_ins(service_name).await;

        remote_service.message_in(message);

        true
    }
    //

    pub async fn get_all_local_service_id_and_name(&self) -> Vec<(i64, String)> {
        let local_service_holder = self.local_service_holder.read().await;

        let mut service_infos = Vec::with_capacity(local_service_holder.service_id_map.len());

        for (service_id, value) in local_service_holder.service_id_map.iter() {
            service_infos.push((*service_id, String::from(value.name())));
        }

        service_infos
    }

    pub async fn get_all_local_service_info(&self, filter_system: bool) -> ServiceInfos {
        let local_service_holder = self.local_service_holder.read().await;

        let mut service_infos = ServiceInfos::default();

        service_infos.services = Vec::with_capacity(local_service_holder.service_id_map.len());

        for (service_id, service) in local_service_holder.service_id_map.iter() {
            if filter_system {
                if service.is_system() {
                    continue;
                }
            }

            let local_service_info = service.get_service_info();

            service_infos.services.push(local_service_info);
        }

        service_infos
    }

    /**
     *  从新分发消息
     */
    async fn redirect_remote_service_message_to_local(&self, service_name: &str) {
        let local_service = self.get_local_service_by_name(service_name).await;

        if local_service.is_none() {
            return;
        }

        let remote_service = self.get_remote_service_by_name(service_name).await;

        if remote_service.is_none() {
            return;
        }

        remote_service
            .unwrap()
            .redirect_message(local_service.unwrap());

        self.remove_remote_service_by_name(service_name).await;
    }
}
